package org.example;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class Server {

    public static void main(String[] args) throws Exception {

        int port = Integer.parseInt(System.getProperty("port", "8888"));
        //开启服务端
        try(var serverChannel = AsynchronousServerSocketChannel.open()){
            serverChannel.bind(new InetSocketAddress("0.0.0.0", port));
            System.out.printf("服务端启动：%d%n",port);
            //客户端连接，用来告知需要连接
            AtomicReference<AsynchronousSocketChannel> client = new AtomicReference<>(null);
            //客户端那边提供的连接队列
            LinkedBlockingQueue<AsynchronousSocketChannel> channelQueue = new LinkedBlockingQueue<>();

            //虚拟线程池
            var executor = Executors.newVirtualThreadPerTaskExecutor();

            while (!Thread.interrupted()){
                try {
                    AsynchronousSocketChannel srcChannel = serverChannel.accept().get();
                    System.out.println("有连接:"+srcChannel);

                    // 提交到虚拟线程池
                    executor.submit(() -> {
                        try {
                            var buffer = ByteBuffer.allocate(1024);
                            var len = srcChannel.read(buffer).get();
                            //切换读状态
                            buffer.flip();

                            if(len < 0) {
                                srcChannel.close();
                            }else if(len == 4){
                                //用一个整形长度判断
                                var code = buffer.getInt();
                                if(code == 100){
                                    //关闭旧连接
                                    if(client.get() != null){
                                        client.get().close();
                                    }
                                    //替换新客户端连接
                                    client.set(srcChannel);
                                }else if (code == 200){
                                    //加入连接队列
                                    channelQueue.offer(srcChannel);
                                }else{
                                    //错误数据关闭连接
                                    srcChannel.close();
                                }
                            }else{
                                //告诉客户端需要连接
                                if(client.get() == null){
                                    //没有客户端连接
                                    srcChannel.close();
                                }else{
                                    //告知客户端需要连接
                                    client.get().write(ByteBuffer.wrap(new byte[]{66})).get();
                                    AsynchronousSocketChannel asc = channelQueue.poll(5, TimeUnit.SECONDS);

                                    if(asc!=null){
                                        try (asc;srcChannel){
                                            while (buffer.remaining()>0){
                                                asc.write(buffer).get();
                                            }

                                            System.out.println(srcChannel);

                                            //开始交换数据
                                            Future<?> taskFuture1 = executor.submit(() -> {
                                                transferData(srcChannel, asc);
                                            });

                                            Future<?> taskFuture2 = executor.submit(() -> {
                                                transferData(asc, srcChannel);
                                            });

                                            taskFuture1.get();
                                            taskFuture2.get();

                                        }catch (Exception e){
                                            e.printStackTrace();
                                        }
                                    }
                                }
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                            if(srcChannel!=null){
                                try {
                                    srcChannel.close();
                                } catch (IOException ex) {
                                    ex.printStackTrace();
                                }
                            }
                        }

                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    private static void transferData(AsynchronousSocketChannel srcChannel, AsynchronousSocketChannel dstChannel) {
        int readLen;
        do{
            try {
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                Future<Integer> read1 = srcChannel.read(buffer);
                readLen = read1.get();
                if(readLen>0){
                    buffer.flip();
                    while (buffer.remaining()>0){
                        Future<Integer> write1 = dstChannel.write(buffer);
                        write1.get();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                break;
            }
        }while (readLen>0);
    }
}
